package engine

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/workflow/definition"
	"github.com/louloulin/dataflare/pkg/workflow/dsl"
)

// ExecutionEngine represents a workflow execution engine
type ExecutionEngine interface {
	// Initialize initializes the engine
	Initialize(config *EngineConfig) error
	// DeployWorkflow deploys a workflow
	DeployWorkflow(workflow *definition.Workflow) error
	// RunWorkflow runs a workflow
	RunWorkflow(workflowID string) error
	// StopWorkflow stops a workflow
	StopWorkflow(workflowID string) error
	// GetWorkflowStatus gets the workflow status
	GetWorkflowStatus(workflowID string) (*WorkflowStatus, error)
	// GetNodeStatus gets the node status
	GetNodeStatus(workflowID string, nodeID string) (*NodeStatus, error)
}

// EngineConfig represents engine configuration
type EngineConfig struct {
	// Type is the engine type
	Type string `json:"type"`
	// StateStore is the state store configuration
	StateStore map[string]interface{} `json:"stateStore,omitempty"`
	// Parallelism is the default parallelism
	Parallelism int `json:"parallelism,omitempty"`
	// Checkpointing contains checkpointing settings
	Checkpointing *CheckpointingConfig `json:"checkpointing,omitempty"`
}

// CheckpointingConfig represents checkpointing settings
type CheckpointingConfig struct {
	// Interval is the checkpointing interval
	Interval time.Duration `json:"interval"`
	// Mode is the checkpointing mode (exactly_once, at_least_once)
	Mode string `json:"mode"`
}

// WorkflowStatus represents workflow status
type WorkflowStatus struct {
	// WorkflowID is the workflow ID
	WorkflowID string `json:"workflowId"`
	// RunID is the run ID
	RunID string `json:"runId"`
	// Status is the workflow status
	Status definition.WorkflowState `json:"status"`
	// StartTime is the start time
	StartTime time.Time `json:"startTime"`
	// EndTime is the end time
	EndTime time.Time `json:"endTime,omitempty"`
	// NodeStates are the node states
	NodeStates map[string]*NodeStatus `json:"nodeStates"`
	// LastUpdated is the last updated time
	LastUpdated time.Time `json:"lastUpdated"`
	// Error is the error message
	Error string `json:"error,omitempty"`
}

// NodeStatus represents node status
type NodeStatus struct {
	// NodeID is the node ID
	NodeID string `json:"nodeId"`
	// Status is the node status
	Status definition.WorkflowState `json:"status"`
	// StartTime is the start time
	StartTime time.Time `json:"startTime"`
	// EndTime is the end time
	EndTime time.Time `json:"endTime,omitempty"`
	// ProcessedRecords is the number of processed records
	ProcessedRecords int64 `json:"processedRecords"`
	// ErrorRecords is the number of error records
	ErrorRecords int64 `json:"errorRecords"`
	// LastUpdated is the last updated time
	LastUpdated time.Time `json:"lastUpdated"`
	// Error is the error message
	Error string `json:"error,omitempty"`
}

// ActorExecutionEngine represents an actor-based execution engine
type ActorExecutionEngine struct {
	// config is the engine configuration
	config *EngineConfig
	// workflows are the deployed workflows
	workflows map[string]*definition.Workflow
	// workflowStatus are the workflow statuses
	workflowStatus map[string]*WorkflowStatus
	// nodeStatus are the node statuses
	nodeStatus map[string]map[string]*NodeStatus
	// mu is the mutex
	mu sync.RWMutex
	// stateManager is the state manager
	stateManager StateManager
	// connectorIntegration is the connector integration
	connectorIntegration *ConnectorIntegration
}

// StateManager represents a state manager
type StateManager interface {
	// SaveWorkflowState saves workflow state
	SaveWorkflowState(workflowID string, runID string, state *WorkflowStatus) error
	// GetWorkflowState gets workflow state
	GetWorkflowState(workflowID string, runID string) (*WorkflowStatus, error)
	// SaveNodeState saves node state
	SaveNodeState(workflowID string, runID string, nodeID string, state *NodeStatus) error
	// GetNodeState gets node state
	GetNodeState(workflowID string, runID string, nodeID string) (*NodeStatus, error)
	// ListRunsByWorkflow lists all workflow runs
	ListRunsByWorkflow(workflowID string) ([]string, error)
}

// NewActorExecutionEngine creates a new actor-based execution engine
func NewActorExecutionEngine() *ActorExecutionEngine {
	return &ActorExecutionEngine{
		workflows:            make(map[string]*definition.Workflow),
		workflowStatus:       make(map[string]*WorkflowStatus),
		nodeStatus:           make(map[string]map[string]*NodeStatus),
		connectorIntegration: NewConnectorIntegration(),
	}
}

// Initialize initializes the engine
func (e *ActorExecutionEngine) Initialize(config *EngineConfig) error {
	e.config = config

	// TODO: Initialize state manager based on config

	return nil
}

// DeployWorkflow deploys a workflow
func (e *ActorExecutionEngine) DeployWorkflow(workflow *definition.Workflow) error {
	e.mu.Lock()
	defer e.mu.Unlock()

	// Check if the workflow already exists
	if _, ok := e.workflows[workflow.Metadata.ID]; ok {
		return fmt.Errorf("workflow %s already deployed", workflow.Metadata.ID)
	}

	// Store the workflow
	e.workflows[workflow.Metadata.ID] = workflow

	// Create workflow status
	e.workflowStatus[workflow.Metadata.ID] = &WorkflowStatus{
		WorkflowID:  workflow.Metadata.ID,
		RunID:       fmt.Sprintf("run-%d", time.Now().UnixNano()),
		Status:      definition.WorkflowStateCreated,
		StartTime:   time.Now(),
		NodeStates:  make(map[string]*NodeStatus),
		LastUpdated: time.Now(),
	}

	// Create node statuses
	e.nodeStatus[workflow.Metadata.ID] = make(map[string]*NodeStatus)
	for stepID := range workflow.Steps {
		e.nodeStatus[workflow.Metadata.ID][stepID] = &NodeStatus{
			NodeID:      stepID,
			Status:      definition.WorkflowStateCreated,
			StartTime:   time.Now(),
			LastUpdated: time.Now(),
		}
	}

	// Save workflow state if state manager is available
	if e.stateManager != nil {
		err := e.stateManager.SaveWorkflowState(
			workflow.Metadata.ID,
			e.workflowStatus[workflow.Metadata.ID].RunID,
			e.workflowStatus[workflow.Metadata.ID],
		)
		if err != nil {
			return fmt.Errorf("failed to save workflow state: %w", err)
		}
	}

	return nil
}

// RunWorkflow runs a workflow
func (e *ActorExecutionEngine) RunWorkflow(workflowID string) error {
	e.mu.Lock()
	defer e.mu.Unlock()

	// Check if the workflow exists
	workflow, ok := e.workflows[workflowID]
	if !ok {
		return fmt.Errorf("workflow %s not found", workflowID)
	}

	// Check if the workflow is already running
	status, ok := e.workflowStatus[workflowID]
	if !ok {
		return fmt.Errorf("workflow status for %s not found", workflowID)
	}
	if status.Status == definition.WorkflowStateRunning {
		return fmt.Errorf("workflow %s is already running", workflowID)
	}

	// Update workflow status
	status.Status = definition.WorkflowStateRunning
	status.LastUpdated = time.Now()

	// Save workflow state if state manager is available
	if e.stateManager != nil {
		err := e.stateManager.SaveWorkflowState(
			workflowID,
			status.RunID,
			status,
		)
		if err != nil {
			return fmt.Errorf("failed to save workflow state: %w", err)
		}
	}

	// Start workflow execution in a separate goroutine
	go e.executeWorkflow(workflow, status.RunID)

	return nil
}

// StopWorkflow stops a workflow
func (e *ActorExecutionEngine) StopWorkflow(workflowID string) error {
	e.mu.Lock()
	defer e.mu.Unlock()

	// Check if the workflow exists
	_, ok := e.workflows[workflowID]
	if !ok {
		return fmt.Errorf("workflow %s not found", workflowID)
	}

	// Check if the workflow is running
	status, ok := e.workflowStatus[workflowID]
	if !ok {
		return fmt.Errorf("workflow status for %s not found", workflowID)
	}
	if status.Status != definition.WorkflowStateRunning {
		return fmt.Errorf("workflow %s is not running", workflowID)
	}

	// Update workflow status
	status.Status = definition.WorkflowStateCancelled
	status.LastUpdated = time.Now()
	status.EndTime = time.Now()

	// Save workflow state if state manager is available
	if e.stateManager != nil {
		err := e.stateManager.SaveWorkflowState(
			workflowID,
			status.RunID,
			status,
		)
		if err != nil {
			return fmt.Errorf("failed to save workflow state: %w", err)
		}
	}

	// Stop connectors
	err := e.connectorIntegration.StopConnectors()
	if err != nil {
		return fmt.Errorf("failed to stop connectors: %w", err)
	}

	return nil
}

// GetWorkflowStatus gets the workflow status
func (e *ActorExecutionEngine) GetWorkflowStatus(workflowID string) (*WorkflowStatus, error) {
	e.mu.RLock()
	defer e.mu.RUnlock()

	// Check if the workflow exists
	_, ok := e.workflows[workflowID]
	if !ok {
		return nil, fmt.Errorf("workflow %s not found", workflowID)
	}

	// Check if the workflow status exists
	status, ok := e.workflowStatus[workflowID]
	if !ok {
		return nil, fmt.Errorf("workflow status for %s not found", workflowID)
	}

	return status, nil
}

// GetNodeStatus gets the node status
func (e *ActorExecutionEngine) GetNodeStatus(workflowID string, nodeID string) (*NodeStatus, error) {
	e.mu.RLock()
	defer e.mu.RUnlock()

	// Check if the workflow exists
	_, ok := e.workflows[workflowID]
	if !ok {
		return nil, fmt.Errorf("workflow %s not found", workflowID)
	}

	// Check if the node status exists
	nodeStatuses, ok := e.nodeStatus[workflowID]
	if !ok {
		return nil, fmt.Errorf("node statuses for workflow %s not found", workflowID)
	}

	status, ok := nodeStatuses[nodeID]
	if !ok {
		return nil, fmt.Errorf("node status for %s in workflow %s not found", nodeID, workflowID)
	}

	return status, nil
}

// executeWorkflow executes a workflow
func (e *ActorExecutionEngine) executeWorkflow(workflow *definition.Workflow, runID string) {
	// Get the start step
	startStepID := workflow.Start
	if startStepID == "" {
		e.updateWorkflowStatus(workflow.Metadata.ID, runID, definition.WorkflowStateFailed, "start step not defined")
		return
	}

	startStep, ok := workflow.Steps[startStepID]
	if !ok {
		e.updateWorkflowStatus(workflow.Metadata.ID, runID, definition.WorkflowStateFailed, fmt.Sprintf("start step %s not found", startStepID))
		return
	}

	// Execute the workflow
	err := e.executeStep(workflow, runID, startStep)
	if err != nil {
		e.updateWorkflowStatus(workflow.Metadata.ID, runID, definition.WorkflowStateFailed, err.Error())
		return
	}

	// Update workflow status
	e.updateWorkflowStatus(workflow.Metadata.ID, runID, definition.WorkflowStateCompleted, "")
}

// executeStep executes a workflow step
func (e *ActorExecutionEngine) executeStep(workflow *definition.Workflow, runID string, step *definition.Step) error {
	// Update node status
	e.updateNodeStatus(workflow.Metadata.ID, runID, step.ID, definition.WorkflowStateRunning, "")

	// Execute the step based on its type
	var err error
	switch step.Config.Type {
	case definition.StepTypeTask:
		err = e.executeTaskStep(workflow, runID, step)
	case definition.StepTypeCondition:
		err = e.executeConditionStep(workflow, runID, step)
	case definition.StepTypeSwitch:
		err = e.executeSwitchStep(workflow, runID, step)
	case definition.StepTypeLoop:
		err = e.executeLoopStep(workflow, runID, step)
	case definition.StepTypeWait:
		err = e.executeWaitStep(workflow, runID, step)
	default:
		err = fmt.Errorf("unsupported step type: %s", step.Config.Type)
	}

	if err != nil {
		e.updateNodeStatus(workflow.Metadata.ID, runID, step.ID, definition.WorkflowStateFailed, err.Error())
		return err
	}

	// Update node status
	e.updateNodeStatus(workflow.Metadata.ID, runID, step.ID, definition.WorkflowStateCompleted, "")

	// Execute the next step if defined
	if step.Next != "" {
		nextStep, ok := workflow.Steps[step.Next]
		if !ok {
			return fmt.Errorf("next step %s not found", step.Next)
		}
		return e.executeStep(workflow, runID, nextStep)
	}

	return nil
}

// executeTaskStep executes a task step
func (e *ActorExecutionEngine) executeTaskStep(workflow *definition.Workflow, runID string, step *definition.Step) error {
	// TODO: Implement task execution
	// Suppress unused parameter warnings
	_ = workflow
	_ = runID
	_ = step
	return nil
}

// executeConditionStep executes a condition step
func (e *ActorExecutionEngine) executeConditionStep(workflow *definition.Workflow, runID string, step *definition.Step) error {
	// TODO: Implement condition execution
	// Suppress unused parameter warnings
	_ = workflow
	_ = runID
	_ = step
	return nil
}

// executeSwitchStep executes a switch step
func (e *ActorExecutionEngine) executeSwitchStep(workflow *definition.Workflow, runID string, step *definition.Step) error {
	// TODO: Implement switch execution
	// Suppress unused parameter warnings
	_ = workflow
	_ = runID
	_ = step
	return nil
}

// executeLoopStep executes a loop step
func (e *ActorExecutionEngine) executeLoopStep(workflow *definition.Workflow, runID string, step *definition.Step) error {
	// TODO: Implement loop execution
	// Suppress unused parameter warnings
	_ = workflow
	_ = runID
	_ = step
	return nil
}

// executeWaitStep executes a wait step
func (e *ActorExecutionEngine) executeWaitStep(workflow *definition.Workflow, runID string, step *definition.Step) error {
	// TODO: Implement wait execution
	// Suppress unused parameter warnings
	_ = workflow
	_ = runID
	_ = step
	return nil
}

// updateWorkflowStatus updates workflow status
func (e *ActorExecutionEngine) updateWorkflowStatus(workflowID string, runID string, status definition.WorkflowState, errorMsg string) {
	e.mu.Lock()
	defer e.mu.Unlock()

	// Update workflow status
	if workflowStatus, ok := e.workflowStatus[workflowID]; ok {
		workflowStatus.Status = status
		workflowStatus.LastUpdated = time.Now()
		workflowStatus.Error = errorMsg

		if status == definition.WorkflowStateCompleted || status == definition.WorkflowStateFailed || status == definition.WorkflowStateCancelled {
			workflowStatus.EndTime = time.Now()
		}

		// Save workflow state if state manager is available
		if e.stateManager != nil {
			err := e.stateManager.SaveWorkflowState(
				workflowID,
				runID,
				workflowStatus,
			)
			if err != nil {
				// Log the error
				fmt.Printf("failed to save workflow state: %v\n", err)
			}
		}
	}
}

// updateNodeStatus updates node status
func (e *ActorExecutionEngine) updateNodeStatus(workflowID string, runID string, nodeID string, status definition.WorkflowState, errorMsg string) {
	e.mu.Lock()
	defer e.mu.Unlock()

	// Update node status
	if nodeStatuses, ok := e.nodeStatus[workflowID]; ok {
		if nodeStatus, ok := nodeStatuses[nodeID]; ok {
			nodeStatus.Status = status
			nodeStatus.LastUpdated = time.Now()
			nodeStatus.Error = errorMsg

			if status == definition.WorkflowStateCompleted || status == definition.WorkflowStateFailed || status == definition.WorkflowStateCancelled {
				nodeStatus.EndTime = time.Now()
			}

			// Save node state if state manager is available
			if e.stateManager != nil {
				err := e.stateManager.SaveNodeState(
					workflowID,
					runID,
					nodeID,
					nodeStatus,
				)
				if err != nil {
					// Log the error
					fmt.Printf("failed to save node state: %v\n", err)
				}
			}
		}
	}
}

// KubernetesExecutionEngine represents a Kubernetes-based execution engine
type KubernetesExecutionEngine struct {
	// TODO: Implement Kubernetes execution engine
}

// ServerlessExecutionEngine represents a serverless execution engine
type ServerlessExecutionEngine struct {
	// TODO: Implement serverless execution engine
}

// NewExecutionEngine creates a new execution engine based on the type
func NewExecutionEngine(engineType string) (ExecutionEngine, error) {
	switch engineType {
	case "actor":
		return NewActorExecutionEngine(), nil
	case "kubernetes":
		return nil, fmt.Errorf("kubernetes execution engine not implemented")
	case "serverless":
		return nil, fmt.Errorf("serverless execution engine not implemented")
	default:
		return nil, fmt.Errorf("unsupported execution engine type: %s", engineType)
	}
}

// DeployWorkflowFromDSL deploys a workflow from a DSL model
func DeployWorkflowFromDSL(engine ExecutionEngine, dslModel *dsl.DSLModel) error {
	// Create a parser
	parser := dsl.NewParser(nil)

	// Convert DSL model to workflow
	workflow, err := parser.ConvertToWorkflow(dslModel)
	if err != nil {
		return fmt.Errorf("failed to convert DSL model to workflow: %w", err)
	}

	// Deploy the workflow
	err = engine.DeployWorkflow(workflow)
	if err != nil {
		return fmt.Errorf("failed to deploy workflow: %w", err)
	}

	// Create connectors if the engine is an actor engine
	if actorEngine, ok := engine.(*ActorExecutionEngine); ok {
		err = actorEngine.connectorIntegration.CreateConnectorsFromDSL(dslModel)
		if err != nil {
			return fmt.Errorf("failed to create connectors: %w", err)
		}
	}

	return nil
}

// RunWorkflowFromDSLFile runs a workflow from a DSL file
func RunWorkflowFromDSLFile(ctx context.Context, engineType string, filePath string) error {
	// Create a parser
	parser := dsl.NewParser(nil)

	// Parse the DSL file
	dslModel, err := parser.ParseFile(filePath)
	if err != nil {
		return fmt.Errorf("failed to parse DSL file: %w", err)
	}

	// Create an execution engine
	engine, err := NewExecutionEngine(engineType)
	if err != nil {
		return fmt.Errorf("failed to create execution engine: %w", err)
	}

	// Initialize the engine
	err = engine.Initialize(&EngineConfig{
		Type:        engineType,
		Parallelism: dslModel.Config.Execution.Parallelism,
	})
	if err != nil {
		return fmt.Errorf("failed to initialize execution engine: %w", err)
	}

	// Deploy the workflow
	err = DeployWorkflowFromDSL(engine, dslModel)
	if err != nil {
		return fmt.Errorf("failed to deploy workflow: %w", err)
	}

	// Run the workflow
	err = engine.RunWorkflow(dslModel.Name)
	if err != nil {
		return fmt.Errorf("failed to run workflow: %w", err)
	}

	return nil
}
